//
//  Task sink - design 3
//  Binds PULL socket to tcp://localhost:5558
//  Collects results from workers via that socket
//
//  10000 tasks
//
package main

import (
	"fmt"
	"os"
	"zmq"
	"time"
)

func zmq_main() os.Error {
	context, err := zmq.Init(1)
	if err != nil {
		return err
	}
	defer context.Close()

	receiver, err := context.Socket(zmq.PULL)
	if err != nil {
		return err
	}
	defer receiver.Close()

	err = receiver.Bind("tcp://*:5558")
	if err != nil {
		return err
	}
	
	controller,err := context.Socket(zmq.PUB)
	if err != nil {
		return err
	}
	defer controller.Close()
	err = controller.Bind("tcp://*:5559")
	if err != nil {
		return err
	}
	
	// Wait for start of batch
	_, err = receiver.Recv(0)
	if err != nil {
		return err
	}

	start_time := time.Nanoseconds()
	for task_nbr := 0; task_nbr < 10000; task_nbr++ {
		_, err = receiver.Recv(0)
		if err != nil {
			return err
		}
		if (task_nbr/10)*10 == task_nbr {
			fmt.Printf(":")
		} else {
			fmt.Printf(".")
		}
	}
	fmt.Println()
	fmt.Printf("Total elapsed time: %d msec\n", (time.Nanoseconds()-start_time)/1e6)
	
	err = controller.Send([]byte("KILL"),0)
	if err != nil {
		return err
	}
	return nil
}

var errch = make(chan os.Error)

func main() {
	fmt.Println("Task sink")
	exit := make(chan bool)
	go func() {
		for {
			err := <-errch
			if err != nil {
				fmt.Println(err)
				os.Exit(1)
			}
		}
	}()
	go func() {
		err := zmq_main()
		if err != nil {
			errch <- err
		} else {	
			exit <- true
		}
	}()
	<-exit
}